#include "thread_pool.h"


static void thread_pool_exit_handler(void *data);
static void *thread_pool_cycle(void *data);
static int_t thread_pool_init_default(thread_pool_t *tpp, char *name);

static uint_t       thread_pool_task_id;   // 任务队列中任务的id

static int debug = 0;

// 初始化线程池
thread_pool_t* thread_pool_init()
{
    int             err;
    pthread_t       tid;    // 线程的句柄
    uint_t          n;
    pthread_attr_t  attr;   // 用来设置线程的属性
	thread_pool_t   *tp = NULL;

	tp = (thread_pool_t *)calloc(1,sizeof(thread_pool_t));   // 内存分配（线程池）

	if(tp == NULL){
	    fprintf(stderr, "thread_pool_init: calloc failed!\n");
	}
	// 初始化线程池中其他的数据变量
	thread_pool_init_default(tp, NULL);
	// 初始化任务队列
    thread_pool_queue_init( &tp->queue );
	// 创建互斥量
    if (thread_mutex_create(&tp->mtx) != OK) {
		free(tp);
        return NULL;
    }
	
	// 创建条件变量
    if (thread_cond_create(&tp->cond) != OK) {
        (void) thread_mutex_destroy(&tp->mtx);
		free(tp);
        return NULL;
    }
	// 初始化 pthread_attr_t  attr;
    err = pthread_attr_init(&attr);
    if (err) {
        fprintf(stderr, "pthread_attr_init() failed, reason: %s\n",strerror(errno));
		free(tp);
        return NULL;
    }
	
	// 设置线程的属性为分离状态   （detached -> 分离 ）
    err = pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
    if (err) {
        fprintf(stderr, "pthread_attr_setdetachstate() failed, reason: %s\n",strerror(errno));
		free(tp);
        return NULL;
    }
	
	// 创建线程
    for ( n = 0 ; n < tp->threads; n++ ) {
		// 参数是一个 tp（线程池）
        err = pthread_create(&tid, &attr, thread_pool_cycle, tp );
        if (err) {
            fprintf(stderr, "pthread_create() failed, reason: %s\n",strerror(errno));
			free(tp);
            return NULL;
        }
    }
	//销毁 pthread_attr_t 变量
    (void) pthread_attr_destroy(&attr);
    return tp;  //  返回线程池指针
}

// 销毁线程池
void thread_pool_destroy( thread_pool_t *tp )
{
    uint_t           n;
    thread_task_t    task;    
    volatile uint_t  lock;   // volatile 修饰 表示该变量的值可能会在任何时候被修改，因此需要特殊对待

    memset( &task, '\0',  sizeof(thread_task_t) );

	// 销毁线程池 ----> 执行一个自杀的任务
    task.handler = thread_pool_exit_handler;
    task.ctx = (void *) &lock;   // 参数
	
	// 销毁线程池 ， 
    for (n = 0; n < tp->threads; n++) {
        lock = 1;
		// 将自杀任务加入线程池的任务队列
        if ( thread_task_post(tp, &task) != OK) {
            return;
        }

        while (lock) {
            sched_yield();  //让出 cpu 的使用权
        }

        //task.event.active = 0;
    }

    (void) thread_cond_destroy(&tp->cond);
    (void) thread_mutex_destroy(&tp->mtx);

	free(tp);
}

static void
thread_pool_exit_handler( void *data )   // 线程销毁执行的任务
{
    uint_t *lock = data;

    *lock = 0;
	
    pthread_exit(0);   // 结束线程
}


thread_task_t *
thread_task_alloc(   size_t size )   // 为每一个任务分配内存 ，size_t size-->所需参数所占的空间大小
{
    thread_task_t  *task;

    task = calloc( 1,sizeof(thread_task_t) + size ); // 任务节点本身的大小 + 参数所占空间的大小
    if (task == NULL) {
        return NULL;
    }
	
    task->ctx = task + 1;  // 参数所在的内存地址

    return task;
}


// 往线程池中投递任务
int_t
thread_task_post( thread_pool_t *tp,  thread_task_t *task )  // 传入线程池指针 ，和任务指针
{
	// 加锁
    if ( thread_mutex_lock(&tp->mtx) != OK) {
        return ERROR;
    }
	
	// 当前的任务数量已经达到最大值 
    if ( tp->waiting >= tp->max_queue ) {
        (void) thread_mutex_unlock(&tp->mtx);   // 解锁
        fprintf(stderr,"thread pool \"%s\" queue overflow: %ld tasks waiting\n",
                      tp->name, tp->waiting);
        return ERROR;
    }
	
    //task->event.active = 1;

    task->id = thread_pool_task_id++;
    task->next = NULL;
	
	//唤醒一个线程
    if ( thread_cond_signal( &tp->cond ) != OK ) {
        (void) thread_mutex_unlock(&tp->mtx);
        return ERROR;
    }
	
    *tp->queue.last = task;           // 将任务链到到任务队列的尾部
    tp->queue.last = &task->next;    // 更新尾部节点的位置

    tp->waiting++;

    (void) thread_mutex_unlock(&tp->mtx);

    if(debug)fprintf(stderr,"task #%lu added to thread pool \"%s\"\n",
                   task->id, tp->name);
	
    return OK;
}

static void *
thread_pool_cycle( void *data )  // 线程循环函数（回调）会不断从线程池任务队列拿取任务执行
{
    thread_pool_t *tp = data;    // 该任务的线程池指针

    int                 err;
    thread_task_t       *task;   // 具体任务

    if( debug )fprintf( stderr,"thread in pool \"%s\" started\n", tp->name );
	
    for (  ; ;  ) {          // 死循环，不断从任务队列拿取任务
        if (thread_mutex_lock(&tp->mtx) != OK) {  // 加锁
            return NULL;
        }
		
        tp->waiting--;   // 等待的任务数 -1 
		
		while ( tp->queue.first == NULL ) {   // 判断队列是否为空  
            if ( thread_cond_wait(&tp->cond, &tp->mtx )    // 等待条件变量的成立
                != OK)
            {
                (void) thread_mutex_unlock(&tp->mtx);
                return NULL;
            }
        }
		
		// 移动头节点
        task = tp->queue.first;
        tp->queue.first = task->next;

        if (tp->queue.first == NULL) {   // 判断是否达到尾部的下一个节点（NULL 空节点）
            tp->queue.last = &tp->queue.first;  
        }
		
        if (thread_mutex_unlock(&tp->mtx) != OK) {  // 解锁
            return NULL;
        }
		
        if(debug) fprintf(stderr,"run task #%lu in thread pool \"%s\"\n",
                       task->id, tp->name);

        task->handler( task->ctx );   // 执行任务

        if(debug) fprintf(stderr,"complete task #%lu in thread pool \"%s\"\n",task->id, tp->name);

        task->next = NULL;

        //notify 
    }
}

static int_t
thread_pool_init_default(thread_pool_t *tpp, char *name)   // 初始化线程池的其他属性
{
	if(tpp)
    {
        tpp->threads = DEFAULT_THREADS_NUM;
        tpp->max_queue = DEFAULT_QUEUE_NUM;
        //strdup 字符串拷贝 , 内部会调用malloc 进行内存分配,传入的参数不能为空
		tpp->name = strdup( name?name:"default" );
        if(debug)fprintf(stderr,
                      "thread_pool_init, name: %s ,threads: %lu max_queue: %ld\n",
                      tpp->name, tpp->threads, tpp->max_queue);

        return OK;
    }

    return ERROR;
}




